package com.aotain.nyx.ipstat;

import java.util.HashMap;
import java.util.Map.Entry;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;

/**
 * 网络源IP流量统计
 * <功能详细描述>
 * 
 * @author  姓名 工号
 * @version  [版本号, 2017年4月6日]
 * @see  [相关类/方法]
 * @since  [产品/模块版本]
 */
public class IPFlowStatReduce implements ReduceFunction<Tuple2<String,IPStatTuple>> {

	/**
	 * serialVersionUID
	 */
	private static final long serialVersionUID = 8188172136381278663L;

	/**
	 * 
	 */
	@Override
	public Tuple2<String, IPStatTuple> reduce(
			Tuple2<String, IPStatTuple> value1,
			Tuple2<String, IPStatTuple> value2) throws Exception {
		// TODO Auto-generated method stub
		Tuple2<String, IPStatTuple> ret = new Tuple2<String, IPStatTuple>();
		
		String dip = value1.f0;
		IPStatTuple col0 = value1.f1;
		IPStatTuple col1 = value2.f1;
		if(col0 == null || col1 == null)
			return null;
		
		IPStatTuple retTuple = new IPStatTuple();
		retTuple.setDestIP(dip);
		
		HashMap<String,Long> mapAcc = new HashMap<String,Long>();
		for(Entry<String,Long> entry: col0.getValueMap().entrySet()) {
			String key = entry.getKey();
			Long value = entry.getValue();
			if(mapAcc.containsKey(key)) {
				mapAcc.put(key, mapAcc.get(key) + value);
			} else {
				mapAcc.put(key, value);
			}
		}
		
		for(Entry<String,Long> entry: col1.getValueMap().entrySet()) {
			String key = entry.getKey();
			Long value = entry.getValue();
			if(mapAcc.containsKey(key)) {
				mapAcc.put(key, mapAcc.get(key) + value);
			} else {
				mapAcc.put(key, value);
			}
		}
		
		retTuple.setValueMap(mapAcc);
		
		/*计算峰值*/
		
		if(col0.getAttackMax() >= col1.getAttackMax()) {
			retTuple.setAttackMax(col0.getAttackMax());
		} else {
			retTuple.setAttackMax(col1.getAttackMax());
		}
		
		ret.setFields(dip, retTuple);
		return ret;
	}

}
